package com.boot.es;

import com.alibaba.fastjson.JSON;
import com.boot.es.model.EsProduct;
import com.boot.es.service.ProductService;
import com.boot.forest.MyService;
import com.boot.utils.Result;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequestBuilder;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.*;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryAction;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import javax.annotation.Resource;
import java.net.InetAddress;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * @author zhangsw
 * @Package com.boot.es
 * @date 2020/12/2 17:57
 */
@Slf4j
public class TransportClientTester {

    private TransportClient client;
    private static final String index_name="test_index";
    private static final String type="product";
    private static final Random random=new Random();
    private BulkProcessor bulkProcessor;

    @Resource
    private MyService myService;

    @Before
    public void setup(){
        try {
            //注意这儿是tcp链接 是9300
            TransportAddress transportAddress = new TransportAddress(InetAddress.getByName("10.27.14.9"), 9300);
            Settings settings = Settings.builder()
                    .put("client.transport.sniff", true)//增加嗅探机制，找到ES集群
                    .put("thread_pool.search.size", 2)//增加线程池个数，暂时设为5
                    .put("cluster.name", "elasticsearch").build();
            client = new PreBuiltTransportClient(settings);
            client.addTransportAddress(transportAddress);
        }catch (Exception e){
            e.printStackTrace();
            log.info("es client 初始化失败");
        }


        BulkProcessor.Listener listener=new BulkProcessor.Listener() {
            //todo beforeBulk会在批量提交之前执行
            public void beforeBulk(long l, org.elasticsearch.action.bulk.BulkRequest bulkRequest) {
                log.info("---尝试操作{}条数据---",bulkRequest.numberOfActions());
            }
            //TODO 第一个afterBulk会在批量成功后执行，可以跟beforeBulk配合计算批量所需时间
            public void afterBulk(long l, org.elasticsearch.action.bulk.BulkRequest bulkRequest, BulkResponse bulkResponse) {
                log.info("---尝试操作{}条数据成功---",bulkRequest.numberOfActions());
            }
            //TODO 第二个afterBulk会在批量失败后执行
            public void afterBulk(long l, org.elasticsearch.action.bulk.BulkRequest bulkRequest, Throwable throwable) {
                log.info("---尝试操作{}条数据失败---",bulkRequest.numberOfActions());
            }

        };

        bulkProcessor = BulkProcessor.builder(client,listener)
                // 1w次请求执行一次bulk
                .setBulkActions(1000)
                // 1gb的数据刷新一次bulk
                .setBulkSize(new ByteSizeValue(1, ByteSizeUnit.MB))
                // 固定5s必须刷新一次
                .setFlushInterval(TimeValue.timeValueSeconds(5))
                // 并发请求数量, 0不并发, 1并发允许执行
                .setConcurrentRequests(1)
                // 设置退避, 100ms后执行, 最大请求3次
                .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3))
                .build();

    }

    @After
    public void cleanup(){
        // 关闭
        try {
            bulkProcessor.awaitClose(10, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void test_save(){
        log.info("-----------single save start---------");
        int id=random.nextInt(10000)+1;
        id=1858;
        EsProduct product=new EsProduct(id,"小米note4","0.8",100.00,new Date(),"家电","小米");
        IndexResponse indexResponse = client.prepareIndex(index_name, type)
                .setSource(JSON.toJSONString(product), XContentType.JSON).get();
        log.info("indexResponse ={}",indexResponse);
        log.info("-----------single save end---------");

    }

    @Test
    public void test_saveId(){
        log.info("-----------single save set id start---------");
        int id=random.nextInt(10000)+1;
        id=1858;
        EsProduct product=new EsProduct(id,"小米note3","0.8",100.00,new Date(),"家电","红米");
        IndexResponse indexResponse = client.prepareIndex(index_name, type,id+"")
                .setSource(JSON.toJSONString(product), XContentType.JSON).get();
        log.info("indexResponse ={}",indexResponse);
        log.info("-----------single save set id end---------");
    }

    /** 普通批量保存 */
    @Test
    public void test_batchsave(){
        log.info("-----------batch save set id start---------");
        Instant startTime=Instant.now();
        BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
        for(int i=0;i<3;i++){

            int id=random.nextInt(10000)+1;
            EsProduct product=new EsProduct(id,"小米note3","0.8",100.00,new Date(),"家电","红米");

            IndexRequestBuilder indexRequestBuilder=client.prepareIndex(index_name, type,id+"")
                    .setSource(JSON.toJSONString(product), XContentType.JSON);
            indexRequestBuilder.request();
            bulkRequestBuilder.add(indexRequestBuilder);
        }

        BulkResponse bulkItemResponses = bulkRequestBuilder.get();
        long millis = ChronoUnit.MILLIS.between(startTime, Instant.now());
        if(bulkItemResponses.hasFailures())
            log.info("bulkItemResponses failures ={}",bulkItemResponses.buildFailureMessage());

        log.info("-----------batch save set id end  millis={}---------",millis);
    }

    /** BulkProcessor */
    @Test
    public void test_batchsave1(){
        log.info("-----------batch save set id start---------");
        Instant startTime=Instant.now();
        for(int i=0;i<3;i++){

            int id=random.nextInt(10000)+1;
            EsProduct product=new EsProduct(id,"小米note3","0.8",100.00,new Date(),"家电","红米");

            IndexRequestBuilder indexRequestBuilder=client.prepareIndex(index_name, type,id+"")
                    .setSource(JSON.toJSONString(product), XContentType.JSON);
            bulkProcessor.add(indexRequestBuilder.request());
        }
//        bulkProcessor.flush();//强刷
        long millis = ChronoUnit.MILLIS.between(startTime, Instant.now());
        log.info("-----------batch save set id end millis={}---------",millis);
    }

    @Test
    public void test_del(){
        log.info("-----------single delete start---------");
        int id=random.nextInt(10000)+1;
        EsProduct product=new EsProduct(id,"小米note3","0.8",100.00,new Date(),"家电","红米");
        IndexResponse indexResponse = client.prepareIndex(index_name, type,id+"")
                .setSource(JSON.toJSONString(product), XContentType.JSON).get();
        log.info("indexResponse ={}",indexResponse);

        DeleteResponse deleteResponse=client.prepareDelete(index_name,type,id+"").get();
        log.info("deleteResponse ={}",deleteResponse);

        log.info("-----------single delete end---------");
    }

    @Test
    public void test_del_ids(){
        log.info("-----------single delete start---------");
        List<String> ids=new ArrayList<>();
        for(int i=0;i<3;i++){
            int id=random.nextInt(10000)+1;
            ids.add(id+"");
            EsProduct product=new EsProduct(id,"小米note3","0.8",100.00,new Date(),"家电","红米");

            IndexRequestBuilder indexRequestBuilder=client.prepareIndex(index_name, type,id+"")
                    .setSource(JSON.toJSONString(product), XContentType.JSON);
            bulkProcessor.add(indexRequestBuilder.request());
        }
        bulkProcessor.flush();//强刷
        String[] idsArray = new String[ids.size()];
        IdsQueryBuilder queryBuilders= QueryBuilders.idsQuery(type)
                .addIds(ids.toArray(idsArray));

        BulkByScrollResponse searchResponse=DeleteByQueryAction.INSTANCE.newRequestBuilder(client)
                .filter(queryBuilders)
                .source(index_name)
                .get();
        log.info("deleteResponse ={}",searchResponse);

        log.info("-----------single delete end---------");
    }

    /**
     * text字段会分词 keyword字段不分词
     * 1、term查询keyword字段   term不会分词。而keyword字段也不分词。需要完全匹配才可。
     * 2、term查询text字段。因为text字段会分词，而term不分词，所以term查询的条件必须是text字段分词后的某一个
     */
    @Test
    public  void termQuery(){
        log.info("-----------single delete start---------");
        SearchResponse searchResponse = client.prepareSearch(index_name).setTypes(type)
                .setQuery(new TermQueryBuilder("title", "小米note3"))
                .setVersion(true)
                .get();

        SearchHits hits = searchResponse.getHits();
        log.info("termQuery totalHits={}",hits.totalHits);
        for(SearchHit hit:hits.getHits()){
            log.info("termQuery hits={}",hit.getSourceAsString());
        }

        log.info("-----------single delete end---------");
    }


    @Test
    public  void boolQuery(){
        log.info("-----------single delete start---------");

        BoolQueryBuilder queryBuilder=QueryBuilders.boolQuery()
                .must(QueryBuilders.rangeQuery("id").gte(31).lte(53))
                .must(QueryBuilders.termQuery("title.keyword","小米note3"));

        SearchResponse searchResponse = client.prepareSearch(index_name).setTypes(type)
                .setQuery(queryBuilder)
                .setFrom(0)
                .setSize(20)
                .get();

        SearchHits hits = searchResponse.getHits();
        log.info("termQuery totalHits={}",hits.totalHits);
        for(SearchHit hit:hits.getHits()){
            log.info("termQuery hits={}",hit.getSourceAsString());
        }

        log.info("-----------single delete end---------");
    }

    @Test
    public void forestTest(){
        myService.testClient();
    }

}
